package com.ninth_group.utils;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.FileSystem;

import java.io.*;
import java.net.URI;

/**
 * HDFS 文件上传工具类
 */
public class HdfsUploader {
    static {
        System.setProperty("hadoop.home.dir", "C:\\dummy\\hadoop");
    }
    private static final String HDFS_URI = "hdfs://master:9000";

    public static void upData(String localFilePath, String hdfsTargetDir) {
        try (InputStream is = HdfsUploader.class.getClassLoader().getResourceAsStream(localFilePath)) {
            if (is == null) throw new RuntimeException("文件未找到: " + localFilePath);

            // 保存为本地临时文件
            Path localPath = saveToTemp(is, localFilePath);

            // 提取文件名
            String fileName = new File(localFilePath).getName();

            // 如果hdfsTargetDir不是以/结尾，则添加之
            if (!hdfsTargetDir.endsWith("/")) {
                hdfsTargetDir += "/";
            }

            // 完整的目标文件路径
            String hdfsTargetPath = hdfsTargetDir + fileName;

            // 配置 HDFS
            Configuration conf = new Configuration();
            conf.set("dfs.client.use.datanode.hostname", "true");
            FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, "root");

            // 上传文件
            Path hdfsPath = new Path(hdfsTargetPath);
            if (fs.exists(hdfsPath)) fs.delete(hdfsPath, true);
            fs.copyFromLocalFile(false, true, localPath, hdfsPath);

            System.out.println("上传成功！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /*public static void upData(String localFilePath, String hdfsTargetPath) {
        try (InputStream is = HdfsUploader.class.getClassLoader().getResourceAsStream(localFilePath)) {
            if (is == null) throw new RuntimeException("文件未找到: " + localFilePath);

            // 保存为本地临时文件
            Path localPath = saveToTemp(is, localFilePath);

            // 配置 HDFS
            Configuration conf = new Configuration();
            conf.set("dfs.client.use.datanode.hostname", "true");
            FileSystem fs = FileSystem.get(new URI(HDFS_URI), conf, "root");

            // 上传文件
            Path hdfsPath = new Path(hdfsTargetPath);
            if (fs.exists(hdfsPath)) fs.delete(hdfsPath, true);
            fs.copyFromLocalFile(false, true, localPath, hdfsPath);

            System.out.println("上传成功！");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }*/

    private static Path saveToTemp(InputStream is, String filename) throws IOException {
        File temp = File.createTempFile("hdfs-temp-", filename);
        try (FileOutputStream fos = new FileOutputStream(temp)) {
            byte[] buf = new byte[1024];
            int len;
            while ((len = is.read(buf)) > 0) {
                fos.write(buf, 0, len);
            }
        }
        return new Path(temp.getAbsolutePath());
    }

}